package rep

import (
    "../pb"
    "../common"
)

func (r *Replica) status() pb.PartitionStatus {
    r.mu.RLock()
    defer r.mu.RUnlock()
    return r.repConfig.Status
}

func (r *Replica) ballot() int64 {
    r.mu.RLock()
    defer r.mu.RUnlock()
    return r.repConfig.Ballot
}

func (r *Replica) updateConfiguration(proposal *pb.PartitionConfiguration) bool {
    //TODO
    return false
}

func (r *Replica) lastCommittedOffset() int64 {
    //TODO
    return 0
}

func (r *Replica) updateConfigurationOnMetaServer(configType pb.ConfigType, node string, config *pb.PartitionConfiguration) {
    config.LastCommittedOffset = r.lastCommittedOffset()
    //some log here
    r.updateLocalConfigurationWithNoBallotChange(pb.PartitionStatus_PS_Inactive)
    r.setInactiveStateTransient(true)
    config.Ballot++
    req := pb.ConfigurationUpdateRequest{
        ConfigType: configType,
        Config:     config,
        Node:       node,
        FromNode:   r.myNode.Addr(""), //??
    }
    if r.primaryState.reconfigurationTask != nil {
        r.primaryState.reconfigurationTask.Cancel(true)
    }
    //TODO find meta
    /*
    _primary_states.reconfiguration_task =
        rpc::call(target,
                  msg,
                  this,
                  [=](error_code err, dsn_message_t reqmsg, dsn_message_t response) {
                      on_update_configuration_on_meta_server_reply(err, reqmsg, response, request);
                  },
                  gpid_to_thread_hash(get_gpid()));
     */
}

func (r *Replica) onUpdateConfigurationOnMetaServerReply(req pb.ConfigurationUpdateRequest, resp pb.ConfigurationUpdateResponse, err error) {

}

func (r *Replica) setInactiveStateTransient(b bool) {
    r.mu.Lock()
    defer r.mu.Unlock()
    r.inactiveIsTransient = b
}

func (r *Replica) updateLocalConfigurationWithNoBallotChange(status pb.PartitionStatus) {

}

func (r *Replica) downgradeToInactiveOnPrimary(proposal *pb.ConfigurationUpdateRequest) {
    if proposal.Config.Ballot != r.ballot() || r.status() != pb.PartitionStatus_PS_Primary {
        return
    }
    if proposal.Node == proposal.Config.Primary {
        proposal.Config.Primary = ""
    } else {
        var secondaries []string
        for _, n := range proposal.Config.Secondaries {
            if n == proposal.Node {
                continue
            }
            secondaries = append(secondaries, n)
        }
    }
    r.update_configuration_on_meta_server(
        pb.ConfigType_CT_DowngradeToInactive, proposal.Node, proposal.Config)
}

func (r *Replica) downgradeToSecondaryOnPrimary(proposal *pb.ConfigurationUpdateRequest) {
    if proposal.Config.Ballot != r.ballot() || r.status() != pb.PartitionStatus_PS_Primary {
        return
    }
    proposal.Config.Secondaries = append(proposal.Config.Secondaries, proposal.Config.Primary)
    proposal.Config.Primary = ""
    r.update_configuration_on_meta_server(
        pb.ConfigType_CT_DowngradeToSecondary, proposal.Node, proposal.Config)
}

func (r *Replica) addPotentialSecondary(proposal *pb.ConfigurationUpdateRequest) {
    if stat := r.status(); stat != pb.PartitionStatus_PS_Primary {
        //ignore add secondary proposal for invalid state, state = stat
        return
    }
    //TODO check if exceeds max replica count
    node := common.Node(proposal.Node)
    //TODO check if node is valid
    signature := r.primaryState.addLearner(node)
    //do group check
    config := r.primaryState.getReplicaConfig(pb.PartitionStatus_PS_PotentialSecondary, signature)
    groupCheckRequest := pb.GroupCheckRequest{
        Node:                proposal.Node,
        Config:              config,
        LastCommittedOffset: r.lastCommittedOffset(),
    }
    //rpc call proposal.Node with groupCheckRequest, just call, one way
}

func (r *Replica) assignPrimary(proposal *pb.ConfigurationUpdateRequest) {
    if r.status() == pb.PartitionStatus_PS_Primary {
        //log invalid assgin primary proposal as the node is in primary
        return
    }
    if proposal.ConfigType == pb.ConfigType_CT_UpgradeToPrimary &&
        (r.status() != pb.PartitionStatus_PS_Secondary || r.secondaryState.checkpointIsRunning()) {
        //log name() invalid upgrade to primary proposal as the node is in status() or during checkpointing
        return
    }
    proposal.Config.Primary = r.myNode.Addr("")
    var secondaries []string
    for _, n := range proposal.Config.Secondaries {
        if n == proposal.Config.Primary {
            continue
        }
        secondaries = append(secondaries, n)
    }
    proposal.Config.Secondaries = secondaries
    r.updateConfigurationOnMetaServer(proposal.ConfigType, proposal.Node, proposal.Config)
}

func (r *Replica) onConfigProposal(proposal *pb.ConfigurationUpdateRequest) {
    //log process config proposal mynode for node
    if proposal.Config.Ballot < r.ballot() {
        //warn on_config_proposal out-dated
        return
    }
    if r.primaryState.isReconfigurationTaskRunning() {
        //dinfo("%s: reconfiguration on the way, skip the incoming proposal", name());
        return
    }
    if proposal.Config.Ballot > r.ballot() {
        if !r.updateConfiguration(proposal.Config) {
            // is closing or update failed
            return
        }
    }
    switch proposal.ConfigType {
    case pb.ConfigType_CT_AssignPrimary:
    case pb.ConfigType_CT_UpgradeToPrimary:
        r.assignPrimary(proposal)
    case pb.ConfigType_CT_AddSecondary:
    case pb.ConfigType_CT_AddSecondaryForLb:
        r.addPotentialSecondary(proposal)
    case pb.ConfigType_CT_DowngradeToSecondary:
        r.downgradeToSecondaryOnPrimary(proposal)
    case pb.ConfigType_CT_DowngradeToInactive:
        r.downgradeToInactiveOnPrimary(proposal)
    default:
        //dassert(false, "invalid config_type, type = %s", enum_to_string(proposal.type));
    }
}
